package org.common.utils.rpc.net.handler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import org.common.utils.net.netty.core.NettyHandler;
import org.common.utils.rpc.core.RpcImplementPool;
import org.common.utils.rpc.entity.Metadata;
import org.common.utils.rpc.entity.RpcRequest;
import org.common.utils.rpc.entity.RpcResponse;
import org.common.utils.thread.ThreadTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

/**
 * RPC服务端处理器
 * @author zhouzhibing
 * @date 2025/11/16 20:14
 */
@ChannelHandler.Sharable
public class RpcServerNettyHandler extends NettyHandler {

    private static final Logger log = LoggerFactory.getLogger(RpcServerNettyHandler.class);
    /**
     * RPC实现对象池
     */
    private final RpcImplementPool implementPool;
    /**
     * 执行器
     */
    private final Executor executor;

    public RpcServerNettyHandler(RpcImplementPool implementPool, Executor executor) {
        this.implementPool = implementPool;
        this.executor = executor;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(executor == null) {
            handlerRequest(ctx , msg);
        } else {
            executor.execute(() -> handlerRequest(ctx , msg));
        }
    }

    /**
     * 读取请求
     * @param ctx 上下文对象
     * @param msg 消息对象
     */
    private void handlerRequest(ChannelHandlerContext ctx, Object msg) {
        RpcRequest request = (RpcRequest)msg;
        log.info("channelRead request = {}", request);

        Metadata metadata = request.getMetadata();
        Class<?> interfaceClass = metadata.getInterfaceClass();
        String methodName = metadata.getMethodName();
        Object[] args = request.getArgs();

        Object result = null;
        String error = null;
        try {
            result = implementPool.invoke(interfaceClass, methodName , args);
        } catch (Exception e) {
            log.error("channelRead error " , e);
            error = ThreadTool.toExceptionString(e);
        } finally {
            doResponse(ctx , request , result , error);
        }
    }

    /**
     * 处理响应
     * @param ctx 上下文对象
     * @param request 请求对象
     * @param result 结果对象
     * @param error 错误信息
     */
    private void doResponse(ChannelHandlerContext ctx, RpcRequest request , Object result, String error) {
        Channel channel = ctx.channel();
        long requestId = request.getRequestId();

        //如果无返回结果
        if(result == null) {
            RpcResponse response = new RpcResponse(requestId, null , error);
            channel.writeAndFlush(response);
            log.info("channelRead finish response = {}" , response);
        } else {
            //如果是CompletableFuture，进行异步处理。
            if(result instanceof CompletableFuture<?> future) {
                future.whenComplete((o, throwable) -> {
                    try {
                        String futureError = ThreadTool.toExceptionString(throwable);
                        doResponse(ctx, request, o, futureError);
                    } catch (Exception e) {
                        log.error("channelRead future execute error " , e);
                    }
                });
            } else {
                //有返回值，直接返回结果
                RpcResponse response = new RpcResponse(requestId, result , error);
                channel.writeAndFlush(response);
                log.info("channelRead finish response = {}" , response);
            }
        }
    }
}

